package cn.haohaowo.config;

import org.apache.kafka.clients.consumer.ConsumerConfig;
import org.apache.kafka.clients.producer.ProducerConfig;
import org.apache.kafka.common.serialization.StringDeserializer;
import org.apache.kafka.common.serialization.StringSerializer;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.beans.factory.annotation.Qualifier;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import org.springframework.kafka.annotation.EnableKafka;
import org.springframework.kafka.config.ConcurrentKafkaListenerContainerFactory;
import org.springframework.kafka.core.*;
import org.springframework.kafka.listener.AbstractMessageListenerContainer;

import java.util.HashMap;
import java.util.Map;

/**
 * Created by luwancai on 18/4/16.
 * DESC:
 */
@Configuration
@EnableKafka
public class SpringKafkaConfig {

    @Value("${kafka.bootstrap.servers:hadoop-senior01.haohaoshop.com:9092}")
    private String bootStrapServers;

    @Value("${kafka.ack.type:all}")
    private String ackType;

    @Value("${kafka.retries.times:3}")
    private String retriesTimes;

    @Value("${kafka.topic.name:test}")
    private String topicName;

    @Value("${kafka.consumer.groupId:test-consumer-group}")
    private String groupId;

    @Autowired
    private ProducerMessageListener producerMessageListener;


    @Bean(name = "haoShopProducerFactory")
    public DefaultKafkaProducerFactory getKafkaProducerFactory() {
        Map<String, Object> producerProp = new HashMap<String, Object>(4);
        producerProp.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, bootStrapServers);
        producerProp.put(ProducerConfig.ACKS_CONFIG, ackType);
        producerProp.put(ProducerConfig.RETRIES_CONFIG, retriesTimes);
        producerProp.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, StringSerializer.class);
        producerProp.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, StringSerializer.class);
        DefaultKafkaProducerFactory kafkaProducerFactory = new DefaultKafkaProducerFactory(producerProp);

        return kafkaProducerFactory;
    }

    @Bean(name = "haoShopKafkaTemplate")
    public KafkaTemplate getKafkaTemplate(@Qualifier("haoShopProducerFactory") ProducerFactory producerFactory) {
        KafkaTemplate kafkaTemplate = new KafkaTemplate(producerFactory, true);
        kafkaTemplate.setDefaultTopic(topicName);
        kafkaTemplate.setProducerListener(producerMessageListener);
        return kafkaTemplate;
    }

    @Bean(name = "haoShopConsumerFactory")
    public DefaultKafkaConsumerFactory getKafkaConsumerFactory() {
        Map<String, Object> consumerProp = new HashMap<String, Object>(4);
        consumerProp.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, bootStrapServers);
        consumerProp.put(ConsumerConfig.ENABLE_AUTO_COMMIT_CONFIG, false);
        consumerProp.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class);
        consumerProp.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class);
        consumerProp.put(ConsumerConfig.GROUP_ID_CONFIG, groupId);
//        consumerProp.put(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG, "earliest");
        DefaultKafkaConsumerFactory kafkaConsumerFactory = new DefaultKafkaConsumerFactory(consumerProp);

        return kafkaConsumerFactory;
    }

    @Bean(name = "kafkaListenerContainerFactory")
    public ConcurrentKafkaListenerContainerFactory getKafkaListenerContainerFactory(
            @Qualifier("haoShopConsumerFactory") ConsumerFactory consumerFactory) {
        ConcurrentKafkaListenerContainerFactory listenerContainerFactory = new ConcurrentKafkaListenerContainerFactory();
        listenerContainerFactory.setConsumerFactory(consumerFactory);
        listenerContainerFactory.setConcurrency(3);
        listenerContainerFactory.getContainerProperties().setPollTimeout(3000);
        listenerContainerFactory.getContainerProperties().setAckMode(AbstractMessageListenerContainer.AckMode.RECORD);

        return listenerContainerFactory;
    }

}
